package com.lnsoft.dqfs.server.handle.threadpools;

import com.lnsoft.dqfs.server.dao.ConnectArg;
import com.lnsoft.dqfs.server.dao.DqfsDao;
import com.lnsoft.dqfs.server.dao.Yc;
import com.lnsoft.dqfs.server.handle.codec.*;
import com.lnsoft.dqfs.server.util.Arrays16;
import com.lnsoft.dqfs.server.util.DeCrypt;
import com.lnsoft.dqfs.server.util.Line2Yc;
import com.lnsoft.dqfs.server.util.MessageUtil;
import com.lnsoft.dqfs.server.util.UpgradePackageCache;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
/**
 *  业务数据处理线程池
 */
public class ServiceInvokePools {
	private static final Logger log = LoggerFactory.getLogger(ServiceInvokePools.class);


    @Autowired
    private DqfsDao dqfsDao;
    @Autowired
    private UpgradePackageCache upgradePackageCache;

    @Value("${updata.path}")
    private String dataPath;

    @Value("${project.is_test}")
    private boolean isTest;


    private final SleepReadArgCommon sleepReadArgCommon = SleepReadArgCommon.getInstance();


    private final ExecutorService pools = Executors.newFixedThreadPool(20);

    private final ExecutorService dataPools = Executors.newFixedThreadPool(10);


    public void clearPools() {
        dataPools.shutdown();
        pools.shutdown();
    }


    public void execute(final ChannelHandlerContext ctx, final ServiceFrame seviceFrame, final Frame frame) {
        pools.execute(new Runnable() {
            @Override
            public void run() {
                int clientName = frame.getClientName();
                int port = frame.getServerPort();


                // 终端请求连接
                if (seviceFrame instanceof ConnectServiceFrame) {
//                    ConnectServiceFrame connectServiceFrame = (ConnectServiceFrame) seviceFrame;


                    String clientId = null;
                    try {
                        clientId = dqfsDao.getIacmId(clientName, port);
                    } catch (Exception e) {
                        log.error("客户端{}:{},验证身份时发生error!!", clientName, port, e);
                    }

                    if (isTest) { // 测试环境 不对终端进行验证
                        if (clientId == null) {
                            clientId = UUID.randomUUID().toString();
                        }
                    }
                    if (clientId == null) {
                        byte[] unConfimFrame = MessageUtil.createUnConfirmFrame(frame.getA1(), frame.getA2());
                        ctx.writeAndFlush(unConfimFrame);
                        log.debug("客户端{}:{},验证失败,返回报文并关闭连接：{}", clientName, port, Arrays16.toString(unConfimFrame));
                        ctx.channel().close();
                        return;
                    }

//                    Channel channel = ctx.channel();
                    // 初始化结束帧后 加入缓存
                    //  channel.attr(AttributeKey.valueOf("clientId")).set(clientId);

                    //  设置终端的数据库对应的id
                    frame.setClientId(clientId);

                    //  构造确认报文并发送
                    byte[] bytes = MessageUtil.createConfirmFrame(frame.getA1(), frame.getA2());

                    //  主站回复确认报文
                    ctx.writeAndFlush(bytes);
                    //  TODO 终端主发切换到主站主发模式，是否要预留时间间隙
                    log.debug("客户端{}:{},注册认证通过，clientId:{}，并返回确认报文：{}", clientName, port, clientId, Arrays16.toString(bytes));

                    //  构造请求报文:请求链路状态
                    byte[] connectBytes = Arrays.copyOf(bytes, 6);
                    connectBytes[1] = 73;
                    connectBytes[4] = (byte) MessageUtil.createCS(connectBytes, 1, 4);

                    //  主站发送请求报文
                    ctx.writeAndFlush(connectBytes);
                    //  设置主站下发状态:使用完整控制域的十进制数值73(fc=9请求链路状态)
                    frame.setProcess(73);
                    //  设置主站的下发FCB反转位
                    frame.setUpServerFcb(true);
                    log.debug("客户端{}:{},主站请求报文<请求链路状态>：{}", clientName, port, Arrays16.toString(connectBytes));

                    //  备份发送的请求报文
                    frame.setSendingFrame(connectBytes);
                } else if (seviceFrame instanceof ResetLinkServiceFrame) {
                    //  发送组装好的复位远方链路报文
                    ResetLinkServiceFrame resetLinkServiceFrame = (ResetLinkServiceFrame) seviceFrame;
                    byte[] bytes = resetLinkServiceFrame.getRespBytes();
                    ctx.writeAndFlush(bytes);
                    log.debug("客户端{}:{},主站请求报文<复位远方链路>：{}", clientName, port, Arrays16.toString(bytes));

                    //  备份发送的请求报文
                    frame.setSendingFrame(bytes);
                    //  设置主站的下发状态
                    frame.setProcess(64);
                    frame.setUpServerFcb(true);

                } else if (seviceFrame instanceof TimeSyncServiceFrame) {
                    // 终端时钟同步
                    // 时针同步 C= 0 1 fcb 1 3(4位)
                    byte c = MessageUtil.createServerC(frame, (byte) 1, (byte) 3);
                    byte[] time = MessageUtil.createTimeNow();
                    ByteBuf byteBuf = Unpooled.buffer();
                    try {
                        byteBuf.writeByte(104).writeByte(17).writeByte(17).writeByte(104)
                                .writeByte(c).writeShortLE(frame.getClientName())
                                .writeByte(103).writeByte(1).writeByte(6).writeShortLE(frame.getClientName())
                                .writeByte(0).writeByte(0).writeBytes(time).writeByte(0).writeByte(22);
                        byte[] bytes = new byte[byteBuf.writerIndex()];

                        byteBuf.readBytes(bytes);
                        bytes[bytes.length - 2] = (byte) MessageUtil.createCS(bytes, 4, bytes.length - 2);
                        ctx.writeAndFlush(bytes);
                        log.debug("客户端{}:{}，主站请求报文<时钟同步>：{}", clientName, port, Arrays16.toString(bytes));

                        frame.setUpServerFcb(true);
                        frame.setProcess(103);
                        //  备份发送的请求报文
                        frame.setSendingFrame(bytes);

                    } finally {
                        byteBuf.release();
                    }
                } else if (seviceFrame instanceof UploadServiceFrame) {
                    //  标志数据上传完成
//                    frame.setFinished(true);
                    // 终端主动上传遥测数据
                    List<Yc> list = null;
                    UploadServiceFrame uploadServiceFrame = (UploadServiceFrame) seviceFrame;
                    long invokeTime = System.currentTimeMillis();

                    // 1.主动上传 文件字节流
                    ByteBuf byteBuf = frame.getByteBuf();
                    int writerIndex = byteBuf.writerIndex();
                    int readerIndex = byteBuf.readerIndex();

                    if (writerIndex - readerIndex > 0) {
                        byte[] temp = new byte[writerIndex - readerIndex];
                        //  将接收的数据写入临时数组
                        byteBuf.readBytes(temp);
                        //  包装数组
                        ByteBuf fileBytes = Unpooled.wrappedBuffer(temp);

                        //  将接收到的数据写入文件
                        log.debug("客户端{}:{},开始写入当前终端上传文件-----------------------------------------", clientName, port);
                        //  记录原始数据的文件名
                        String fileName = dataPath + clientName + "_" + port + "_" + LocalDateTime.now().toString().replaceAll(":", "-");
                        RandomAccessFile randomAccessFile = null;
                        try {
                            //  打开文件
                            randomAccessFile = new RandomAccessFile(fileName, "rw");
                            //  将接收的数据写入到文件
                            fileBytes.readBytes(randomAccessFile.getChannel(), 0, writerIndex - readerIndex);

                        } catch (Exception e) {
                            log.error("客户端{}:{},file write error:{}", clientName, port, e);
                        } finally {
                            byteBuf.clear();
                            if (randomAccessFile != null) {
                                try {
                                    randomAccessFile.close();
                                } catch (IOException e) {
                                    log.error("客户端{}:{},文件：{}关闭异常", clientName, port, fileName);
                                }
                            }
                        }

                        //  2.将数据写入数据库

//                        ByteArrayOutputStream outputStream = new ByteArrayOutputStream(temp);
//                        inputStream.;

                        //  对字节数组添加一个‘具有处理字节数组的方法’的壳
                        ByteBuf byteBuf1 = Unpooled.wrappedBuffer(temp);


                        //  检查是否需要解密
                        boolean flag = false;
                        flag = dqfsDao.hasEncrypted(frame.getClientId());

                        if (flag) { //  如果需要解密
                            log.info("客户端{}:{},终端需要解密", clientName, port);
                            list = new DeCrypt().encryptedBytes2YcList(clientName, byteBuf1);
                        } else {    //  如果不加密
                        	list = Line2Yc.bytes2YcList(clientName, byteBuf1);
                        }

                        //  清理ByteBuff
                        byteBuf1.clear();
                        byteBuf1.release();

                    }
                    log.debug("客户端{}:{}，主动文件上传写入读取操作耗时:{},写入大小:{}", clientName, port, (System.currentTimeMillis() - invokeTime), writerIndex);
                    if (uploadServiceFrame.isReleaseFlag()) {
                        // 客户端处于断开状态 需要对bytebuf做清理操作
                        //  TODO:此处不应该调用释放内存方法，应为系统自动回收
//                        byteBuf.release();
                    } else {

                        if (list != null && list.size() > 0) {
                            // TODO 只能是处理完昨天数据的上传 才可以做升级 或 修改参数操作
                            if (list.get(0).getCjsj().
                                    startsWith(LocalDateTime.now().minusDays(1).
                                            format(DateTimeFormatter.ofPattern("yyyyMMdd")))) {
                                // 执行升级操作
                                // 获取当前版本号
                                byte[] readArgFrame = MessageUtil.createReadArgFrame(frame, (byte) 5, (byte) 128);
                                // 读取终端当前版本号 入队 等待异步发送
                                //  TODO 确认此处是否要优化
//                                sleepReadArgCommon.push(SleepReadArgCommon.ReadArgCls.builder().ctx(ctx).frame(frame).time(new java.util.Date().getTime()).readArgFrame(readArgFrame).build());
                                ctx.writeAndFlush(readArgFrame);
                                log.debug("客户端{}:{}，主站请求报文<读取版本号>:{}", clientName, port, Arrays16.toString(readArgFrame));

                                //  读参数
                                frame.setProcess(202);
                                frame.setUpServerFcb(true);
                                //  备份发送的请求报文
                                frame.setSendingFrame(readArgFrame);
                            }
                        }
                    }

                    if (list != null) {
                        final List<Yc> ycList = list;
                        dataPools.submit(new Runnable() {
                            @Override
                            public void run() {
                                // 批量入库
                                dqfsDao.saveYcList(clientName, port, frame.getClientId(), ycList);
                                dqfsDao.updateIacmDownTime(frame.getClientId());
                            }
                        });

                    }
                } else if (seviceFrame instanceof UpgradePackageServiceFrame) {
                    UpgradePackageServiceFrame upgradePackageServiceFrame = (UpgradePackageServiceFrame) seviceFrame;
                    int type = upgradePackageServiceFrame.getType();
                    if (type == 0) { // 写文件激活
                        byte[] writeActivateFrame = MessageUtil.createWirteActivateFrame(frame, upgradePackageCache.getFileName(), upgradePackageCache.getPackageSize());
                        ctx.writeAndFlush(writeActivateFrame);
                        log.debug("客户端{}:{}，主站请求报文<请求写文件激活指令>:{}", clientName, port, Arrays16.toString(writeActivateFrame));

                        frame.setProcess(210);
                        frame.setUpServerFcb(true);
                        //  备份主站发送的请求报文
                        frame.setSendingFrame(writeActivateFrame);

                    } else if (type == 1) {     //  写文件过程中
                        log.info("客户端{}:{},线程池收到写文件信号", clientName, port);
                        byte[] packageBytes = MessageUtil.createUpgradePackageFrame(frame, upgradePackageCache);
                        log.debug("客户端{}:{},主站请求报文<下发写文件数据>:{}", clientName, port, Arrays16.toString(packageBytes));
                        ctx.writeAndFlush(packageBytes);

                        frame.setProcess(2100);
                        frame.setUpServerFcb(true);
                        //  备份主站发送的请求报文
                        frame.setSendingFrame(packageBytes);
                    }

                } else if (seviceFrame instanceof ResetServiceFrame) {
                    byte[] resetFrame = MessageUtil.createResetFrame(frame);
                    ctx.writeAndFlush(resetFrame);
                    log.debug("客户端{}:{}，主站请求报文<终端复位进程指令>:{}", clientName, port, Arrays16.toString(resetFrame));

                    frame.setProcess(105);
                    frame.setUpServerFcb(true);
                    // 备份主站发送的请求报文
                    frame.setSendingFrame(resetFrame);

                } else if (seviceFrame instanceof UpConnectArgServiceFrame) {
                    // ip port 修改
                    ConnectArg connectArg = dqfsDao.getConnectArg(frame.getClientId());
                    if (connectArg != null) {
                        log.debug("客户端{}:{}，参数修改开始执行",clientName, port);
                        byte[] argFrame = MessageUtil.createArgFrame(frame, connectArg);
                        ctx.writeAndFlush(argFrame);

                        //  设置流程标志:参数预置
                        frame.setProcess(203);
                        frame.setUpServerFcb(true);

                        //  备份主站的请求报文
                        frame.setSendingFrame(argFrame);

                        frame.setArgId(connectArg.getArgId());
                        log.debug("客户端{}:{}，服务端请求参数预置,当前index：{},报文:{}", clientName, port, frame.getUpgradePackageindex(), Arrays16.toString(argFrame));
                    }

                } else if (seviceFrame instanceof ArgSolidifyServiceFrame) {
                    byte[] bytes = MessageUtil.createArgSolidifyFrame(frame);
                    ctx.writeAndFlush(bytes);
                    log.debug("客户端{}:{}，主站请求报文<参数固化指令>:{}", clientName, port, Arrays16.toString(bytes));

                    //  设置流程标志:参数固化
                    frame.setProcess(2030);
                    frame.setUpServerFcb(true);

                    //  备份主站发送的请求报文
                    frame.setSendingFrame(bytes);

                }
            }
        });
    }

    /**
     * poolsInvokeDeCrypt 定时任务方法：使用线程池解密加密报文文件，并将解密的数据进行入库
     *
     * */
    public void poolsInvokeDeCrypt(String fileDirPath) {
        log.debug("********定时任务开始启动：解密指定目录下的加密报文文件并入库**********");
        pools.execute(new Runnable() {
            @Override
            public void run() {
                //	扫描指定路径下的文件
                File dir = new File(fileDirPath);
                File[] files = dir.listFiles();
                try{
                	for (File file : files) {
                    	//	1.过滤应当本次读取的文件
                		//	TODO:
                    	//	2.查询本次读取文件对应的终端
                		String fileName = file.getName();
                    	log.info("读取到的文件名称:{}", fileName);
                    	/**
                         * 筛出当日上传的加密文件
                         * */
                    	if (!fileName.contains(LocalDateTime.now().minusDays(1).
                                format(DateTimeFormatter.ofPattern("yyyy-MM-dd")))){
                    	    continue;
                        }
                        int clientName = Integer.valueOf(fileName.substring(0, 4));
                        log.info("终端号:{}",clientName);
                        String clientId = dqfsDao.getIacmId(clientName, 2504);
                        log.info("终端ID:{}",clientId);

                        //	3.读取文件数据
                        RandomAccessFile randomAccessFile = new RandomAccessFile(file.getName(),"rw");
                        long len = randomAccessFile.length();
                        byte[] temp = new byte[(int)len];
                        randomAccessFile.readFully(temp);
                        //	读取数据之后关闭文件
                        try {
                            randomAccessFile.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        //	使用ByteBuf进行包装byte[]
                        ByteBuf byteBuf = Unpooled.wrappedBuffer(temp);
                        //	4.解析文件数据为实体类
                        List<Yc> list = Line2Yc.bytes2YcList(clientName, byteBuf);
                        if(list == null) {
                        	continue;
                        }

                        //	推入线程池进行入库
                        dataPools.submit(new Runnable() {
                            @Override
                            public void run() {
                                // 批量入库
                                dqfsDao.saveYcList(clientName, 2504, clientId, list);
                                dqfsDao.updateIacmDownTime(clientId);
                            }
                        });
                    }
                }catch(Exception e) {
                	e.printStackTrace();
                }
            }
        });
    }

}
